package org.niit.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/*
 有状态转换： 就是历史数据进行累计统计

 */
object SparkStreaming_06 {
  def main(args: Array[String]): Unit = {
     //准备环境
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("spark")
    val ssc = new StreamingContext(sparkConf, Seconds(3))
    ssc.sparkContext.setLogLevel("ERROR")

    //设置检查点，对历史数据（出现的次数）进行落盘
    ssc.checkpoint("BD2")

    //监听本地套接字，端口号 9999
    val lines: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)

    val flatMap: DStream[String] = lines.flatMap(_.split(" "))
    val map: DStream[(String, Int)] = flatMap.map((_, 1))
    /*
     第一个参数：表示相同Key的数据
     第二个参数：表示缓冲区中相同Key的数据.。保存上一次的统计结果
     */
    val state = map.updateStateByKey(
      (seq:Seq[Int],buff:Option[Int])=>{
         val count =buff.getOrElse(0) + seq.sum  //缓冲区中的value + 本次序列中次数的和、
        Option(count)
      }
    )

    state.print()

    ssc.start()
    ssc.awaitTermination()


  }
}
